大数据Spark生态系统 修仙之道 Spark Blog

2019-05-15 Docs Language:简体中文 & English Programing Spark Website:www.geekparkhub.com OpenSource GitHub repo size in bytes GeekDeveloper:JEEP-711 Github:github.com/geekparkhub Gitee:gitee.com/geekparkhub

🐘 Spark Technology 修仙之道 金仙道果 🐘

Alt text

🔥 1. Spark 基础 🔥

1.1 Spark 概述

1.1.1 Spark 模块

enter image description here

1.1.2 Spark 特点

1.1.3 Spark 应用场景

1.2 Spark 部署

解压spark-2.1.1-bin-hadoop2.7.tgz

[root@systemhub511 software]# tar -zxvf spark-2.1.1-bin-hadoop2.7.tgz -C /opt/module/

重命名spark-2.1.1-bin-hadoop2.7

[root@systemhub511 module]# mv spark-2.1.1-bin-hadoop2.7/ spark

1.3 Spark 运行模式

💥 1.3.1 Loacl Mode 💥

1.3.1.1 Loacl Mode 概述
1.3.1.2 (求π) & (WordCount) & (本地调试) 官方案例
bin/spark-submit \

--class <main-class>

--master <master-url> \

--deploy-mode <deploy-mode> \

--conf <key>=<value> \

... # other options

<application-jar> \

[application-arguments]

bin/spark-submit \

--class org.apache.spark.examples.SparkPi \

--executor-memory 1G \

--total-executor-cores 1 \

./examples/jars/spark-examples_2.11-2.1.1.jar \

100

[root@systemhub511 spark]# bin/spark-submit \

> --class org.apache.spark.examples.SparkPi \

> --executor-memory 1G \

> --total-executor-cores 1 \

> ./examples/jars/spark-examples_2.11-2.1.1.jar \

> 100

INFO DAGScheduler: Job 0 finished: reduce at SparkPi.scala:38, took 3.059446 s

Pi is roughly 3.1411463141146316

[root@systemhub511 spark]# bin/spark-shell

Spark context Web UI available at http://systemhub511:4040

Spark context available as 'sc' (master = local[*], app id = local-1558677071165).

Spark session available as 'spark'.

Welcome to

      ____              __

     / __/__  ___ _____/ /__

    _\ \/ _ \/ _ `/ __/  '_/

   /___/ .__/\_,_/_/ /_/\_\   version 2.1.1

      /_/



Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_162)

Type in expressions to have them evaluated.

Type :help for more information.



scala>

enter image description here

[root@systemhub511 spark]# mkdir -p input/wordcount

[root@systemhub511 spark]# cd input/wordcount/

[root@systemhub511 wordcount]# vim wordcount_001.txt

hadoop spark hive

hadoop spark hadoop

hbase flume hive

scala java oozie

scala> sc.textFile("/opt/module/spark/input/wordcount/wordcount_001.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

res0: Array[(String, Int)] = Array((scala,1), (spark,2), (hive,2), (hadoop,3), (oozie,1), (flume,1), (java,1), (hbase,1))



scala> 

scala> sc.textFile("/opt/module/spark/input/wordcount/wordcount_001.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile("./output/wordcount/")

[root@systemhub511 spark]# cd output/wordcount/

[root@systemhub511 wordcount]# ll

total 4

-rw-r--r--. 1 root root 79 May 24 14:48 part-00000

-rw-r--r--. 1 root root  0 May 24 14:48 _SUCCESS

[root@systemhub511 wordcount]# cat part-00000 

(scala,1)

(spark,2)

(hive,2)

(hadoop,3)

(oozie,1)

(flume,1)

(java,1)

(hbase,1)

[root@systemhub511 wordcount]#
1.3.1.3 提交流程

enter image description here

1.3.1.4 数据流程
参数列表 参数描述
textFile("input") 读取本地文件input文件夹数据
flatMap(_.split(" ")) 压平操作,按照空格分割符将一行数据映射成一个个单词
map((_,1)) 对每一个元素操作,将单词映射为元组
reduceByKey(_+_) 按照key将值进行聚合相加
collect 将数据收集到Driver端展示

enter image description here

💥 1.3.2 Standalone Mode 💥

1.3.2.1 Standalone Mode 概述

enter image description here

1.3.2.2 StandaloneMode QuickStart
[root@systemhub511 spark]# cd conf/

[root@systemhub511 conf]# mv slaves.template slaves

[root@systemhub511 conf]# mv spark-env.sh.template spark-env.sh

#

# Licensed to the Apache Software Foundation (ASF) under one or more

# contributor license agreements.  See the NOTICE file distributed with

# this work for additional information regarding copyright ownership.

# The ASF licenses this file to You under the Apache License, Version 2.0

# (the "License"); you may not use this file except in compliance with

# the License.  You may obtain a copy of the License at

#

#    http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

#



# A Spark Worker will be started on each of the machines listed below.

systemhub511

systemhub611

systemhub711

# Options for the daemons used in the standalone deploy mode

SPARK_MASTER_HOST=systemhub511

SPARK_MASTER_PORT=7077

[root@systemhub511 module]# scp -r spark/ root@systemhub611:/opt/module/

[root@systemhub511 module]# scp -r spark/ root@systemhub711:/opt/module/

[root@systemhub511 spark]# sbin/start-all.sh

starting org.apache.spark.deploy.master.Master, logging to /opt/module/spark/logs/spark-root-org.apache.spark.deploy.master.Master-1-systemhub511.out

systemhub711: starting org.apache.spark.deploy.worker.Worker, logging to /opt/module/spark/logs/spark-root-org.apache.spark.deploy.worker.Worker-1-systemhub711.out

systemhub611: starting org.apache.spark.deploy.worker.Worker, logging to /opt/module/spark/logs/spark-root-org.apache.spark.deploy.worker.Worker-1-systemhub611.out

systemhub511: starting org.apache.spark.deploy.worker.Worker, logging to /opt/module/spark/logs/spark-root-org.apache.spark.deploy.worker.Worker-1-systemhub511.out

[root@systemhub511 spark]# 

[root@systemhub511 spark]# jps.sh

================        root@systemhub511 All Processes         ===========

30651 org.apache.spark.deploy.worker.Worker

30443 org.apache.spark.deploy.master.Master

813 sun.tools.jps.Jps

================        root@systemhub611 All Processes         ===========

10369 org.apache.spark.deploy.worker.Worker

11777 sun.tools.jps.Jps

================        root@systemhub711 All Processes         ===========

8960 org.apache.spark.deploy.worker.Worker

10364 sun.tools.jps.Jps

[root@systemhub511 spark]# 

bin/spark-submit \

--class org.apache.spark.examples.SparkPi \

--master spark://systemhub511:7077 \

--executor-memory 1G \

--total-executor-cores 1 \

./examples/jars/spark-examples_2.11-2.1.1.jar \

100

[root@systemhub511 spark]# bin/spark-submit \

> --class org.apache.spark.examples.SparkPi \

> --master spark://systemhub511:7077 \

> --executor-memory 1G \

> --total-executor-cores 1 \

> ./examples/jars/spark-examples_2.11-2.1.1.jar \

> 100

INFO DAGScheduler: Job 0 finished: reduce at SparkPi.scala:38, took 6.478381 s

Pi is roughly 3.1405883140588315

[root@systemhub511 spark]# bin/spark-shell --master spark://systemhub511:7077



Spark context Web UI available at http://systemhub511:4040

Spark context available as 'sc' (master = spark://systemhub511:7077, app id = app-20190524174512-0001).

Spark session available as 'spark'.

Welcome to

      ____              __

     / __/__  ___ _____/ /__

    _\ \/ _ \/ _ `/ __/  '_/

   /___/ .__/\_,_/_/ /_/\_\   version 2.1.1

      /_/



Using Scala version 2.11.8 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_162)

Type in expressions to have them evaluated.

Type :help for more information.



scala> sc.textFile("/opt/module/spark/input/wordcount/wordcount_001.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

res0: Array[(String, Int)] = Array((scala,1), (hive,2), (oozie,1), (java,1), (spark,2), (hadoop,3), (flume,1), (hbase,1))



scala>

enter image description here

[root@systemhub511 conf]# mv spark-defaults.conf.template spark-defaults.conf

spark.master                     spark://systemhub511:7077

spark.eventLog.enabled           true

spark.eventLog.dir               hdfs://systemhub511:9000/directory

export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=18080 -Dspark.history.retainedApplications=30 -Dspark.history.fs.logDirectory=hdfs://systemhub511:9000/directory"

spark.eventLog.dir：Application在运行过程中所有信息均记录在该属性指定的路径下.



spark.history.ui.port=18080 WEBUI访问端口号为18080



spark.history.fs.logDirectory=hdfs://systemhub511:9000/directory 配置了该属性后,在start-history-server.sh时就无需再显示指定路径,Spark History Server只展示该指定路径下信息.



spark.history.retainedApplications=30 指定保存Application历史记录个数,如果超过这个值,旧应用程序信息将被删除,这个是内存中应用数,而不是页面上显示应用数.

[root@systemhub511 module]# scp -r spark/ root@systemhub611:/opt/module/

[root@systemhub511 module]# scp -r spark/ root@systemhub711:/opt/module/

[root@systemhub511 hadoop]# sbin/start-dfs.sh

[root@systemhub511 spark]# hadoop fs -mkdir /directory

[root@systemhub511 spark]# sbin/start-all.sh

[root@systemhub511 spark]# sbin/start-history-server.sh

[root@systemhub511 spark]# bin/spark-shell --master spark://systemhub511:7077

sc.textFile("/opt/module/spark/input/wordcount/wordcount_001.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

enter image description here

1.3.2.3 Spark HA 高可用

enter image description here

# SPARK_MASTER_HOST=systemhub511

# SPARK_MASTER_PORT=7077

export SPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER -Dspark.deploy.zookeeper.url=systemhub511,systemhub611,systemhub711 -Dspark.deploy.zookeeper.dir=/spark"

export SPARK_HISTORY_OPTS="-Dspark.history.ui.port=18080 -Dspark.history.retainedApplications=30 -Dspark.history.fs.logDirectory=hdfs://systemhub511:9000/directory"

[root@systemhub511 module]# scp -r spark/ root@systemhub611:/opt/module/

[root@systemhub511 module]# scp -r spark/ root@systemhub711:/opt/module/

[root@systemhub511 spark]# /opt/module/hadoop/sbin/start-dfs.sh

[root@systemhub511 spark]# /opt/module/zookeeper/bin/zkServer.sh start

[root@systemhub611 ~]# /opt/module/zookeeper/bin/zkServer.sh start

[root@systemhub711 ~]# /opt/module/zookeeper/bin/zkServer.sh start

[root@systemhub511 spark]# sbin/start-all.sh

[root@systemhub611 ~]# /opt/module/spark/sbin/start-master.sh

[root@systemhub511 spark]# bin/spark-shell --master spark://systemhub511:7077,systemhub611:7077

http://systemhub511:8080 | systemhub511节点状态为ALIVE
http://systemhub611:8080 | systemhub611节点状态为STANDBY

enter image description here

[root@systemhub511 spark]# jps.sh

================        root@systemhub511 All Processes         ===========

32242 org.apache.hadoop.hdfs.server.namenode.NameNode

11206 org.apache.spark.deploy.master.Master

11368 org.apache.spark.deploy.worker.Worker

9705 org.apache.zookeeper.server.quorum.QuorumPeerMain

32444 org.apache.hadoop.hdfs.server.datanode.DataNode

5228 sun.tools.jps.Jps

================        root@systemhub611 All Processes         ===========

9157 org.apache.spark.deploy.master.Master

8901 org.apache.spark.deploy.worker.Worker

2822 sun.tools.jps.Jps

30214 org.apache.hadoop.hdfs.server.datanode.DataNode

7495 org.apache.zookeeper.server.quorum.QuorumPeerMain

================        root@systemhub711 All Processes         ===========

5312 org.apache.spark.deploy.worker.Worker

31568 sun.tools.jps.Jps

26869 org.apache.hadoop.hdfs.server.namenode.SecondaryNameNode

26647 org.apache.hadoop.hdfs.server.datanode.DataNode

4014 org.apache.zookeeper.server.quorum.QuorumPeerMain

[root@systemhub511 spark]# 

[root@systemhub511 spark]# kill -9 11206

enter image description here

💥 1.3.3 Yarn Mode 💥

1.3.3.1 Yarn Mode 概述

enter image description here

1.3.3.2 YarnMode QuickStart
YARN_CONF_DIR=/opt/module/hadoop/etc/hadoop

spark.master                     spark://systemhub511:7077

spark.eventLog.enabled           true

spark.eventLog.dir               hdfs://systemhub511:9000/directory

spark.yarn.historyServer.address=systemhub511:18080

spark.history.ui.port=18080

<!--是否启动一个线程检查每个任务正使用的物理内存量,如果任务超出分配值,则直接将其杀掉,默认是true -->

<property>

  <name>yarn.nodemanager.pmem-check-enabled</name>

  <value>false</value>

</property>



<!--是否启动一个线程检查每个任务正使用的虚拟内存量,如果任务超出分配值,则直接将其杀掉,默认是true-->

<property>

  <name>yarn.nodemanager.vmem-check-enabled</name>

  <value>false</value>

</property>

[root@systemhub511 module]# scp -r spark/ root@systemhub611:/opt/module/

[root@systemhub511 module]# scp -r spark/ root@systemhub711:/opt/module/

bin/spark-submit \ 

--class org.apache.spark.examples.SparkPi \ 

--master yarn \ 

--deploy-mode client \ 

./examples/jars/spark-examples_2.11-2.1.1.jar\ 

100

💥 1.3.4 Mesos Mode 💥

1.3.4.1 Mesos Mode 概述

💥 1.3.5 运行模式对比 💥

模式 集群数量 集群进程 所属者
Loacl Mode 1 无 Spark
Standalone Mode 3 Master & Worker Spark
Yarn Mode 1 Yarn & HDFS Hadoop

💥 1.3.6 WordCount 实例 💥

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <modelVersion>4.0.0</modelVersion>



    <groupId>com.geekparkhub.core.spark</groupId>

    <artifactId>spark_server</artifactId>

    <packaging>pom</packaging>

    <version>1.0-SNAPSHOT</version>



    <modules>

        <module>spark-common</module>

    </modules>



    <dependencies>

        <dependency>

            <groupId>org.apache.spark</groupId>

            <artifactId>spark-core_2.11</artifactId>

            <version>2.1.1</version>

        </dependency>

    </dependencies>



</project>

<?xml version="1.0" encoding="UTF-8"?>

<project xmlns="http://maven.apache.org/POM/4.0.0"

         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"

         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

    <parent>

        <artifactId>spark_server</artifactId>

        <groupId>com.geekparkhub.core.spark</groupId>

        <version>1.0-SNAPSHOT</version>

    </parent>

    <modelVersion>4.0.0</modelVersion>



    <artifactId>spark-common</artifactId>



    <build>

        <finalName>WordCount</finalName>

        <plugins>

            <plugin>

                <groupId>net.alchim31.maven</groupId>

                <artifactId>scala-maven-plugin</artifactId>

                <version>3.2.2</version>

                <executions>

                    <execution>

                        <goals>

                            <goal>compile</goal>

                            <goal>testCompile</goal>

                        </goals>

                    </execution>

                </executions>

            </plugin>

        </plugins>

    </build>



</project>

package com.geekparkhub.core.spark.application.wordcount



import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * WordCountApplication

  * <p>

  */



object WordCount {

  def main(args: Array[String]): Unit = {



    /**

      * Create SparkConf

      * 创建 SparkConf

      */

    val sparkConf = new SparkConf().setMaster(args(0)).setAppName("WordCountApplication")



    /**

      * Create SparkContext

      * 创建 SparkContext

      */

    val sc = new SparkContext()



    /**

      * Read file

      * 读取文件

      */

    val line: RDD[String] = sc.textFile(args(1))



    /**

      * To flatten

      * 压平

      */

    val word: RDD[String] = line.flatMap(_.split(" "))



    /**

      * Word conversion dual group

      * 单词转换二元组

      */

    val wordAndOne: RDD[(String, Int)] = word.map((_, 1))



    /**

      * Count the total number of words

      * 统计单词总数

      */

    val wordCount: RDD[(String, Int)] = wordAndOne.reduceByKey(_+_)



    /**

      * Write out the file

      * 写出文件

      */

    wordCount.saveAsTextFile(args(2))



    /**

      * Close resource

      * 关闭资源

      */

    sc.stop()

  }

}

[root@systemhub511 ~]# hadoop fs -mkdir -p /core_flow/spark/input/wordcount

hadoop fs -put /opt/module/spark/input/wordcount/wordcount_001.txt /core_flow/spark/input/wordcount

bin/spark-submit \

--class com.geekparkhub.core.spark.application.wordcount.WordCount \

--master yarn \

./lib_jar/WordCount.jar yarn \

/core_flow/spark/input/wordcount/wordcount_001.txt \

/core_flow/spark/output/wordcount

[root@systemhub511 spark]# hadoop fs -ls -R /core_flow/spark/output/wordcount/

SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]

-rw-r--r--   3 root supergroup /core_flow/spark/output/wordcount/_SUCCESS

-rw-r--r--   3 root supergroup /core_flow/spark/output/wordcount/part-00000

-rw-r--r--   3 root supergroup /core_flow/spark/output/wordcount/part-00001

[root@systemhub511 spark]# 

[root@systemhub511 spark]# hadoop fs -cat /core_flow/spark/output/wordcount/part-00000

SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]

(scala,1)

(hive,2)

(oozie,1)

(java,1)

[root@systemhub511 spark]# 

[root@systemhub511 spark]# hadoop fs -cat /core_flow/spark/output/wordcount/part-00001

SLF4J: Actual binding is of type [org.slf4j.impl.Log4jLoggerFactory]

(spark,2)

(hadoop,3)

(flume,1)

(hbase,1)

[root@systemhub511 spark]#

🔥 1.3 Spark Core 🔥

1.3.1 RDD 概述

1.3.1.1 什么是RDD

RDD (Resilient Distributed Dataset)弹性分布式数据集是Spark中最基本数据抽象,代码中是一个抽象类,它代表一个弹性/不可变/可分区/里面的元素可并行计算的集合.

1.3.1.2 RDD 属性
 * Internally, each RDD is characterized by five main properties:

 *

 *  - 1. A list of partitions

 *  - 2. A function for computing each split

 *  - 3. A list of dependencies on other RDDs

 *  - 4. Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)

 *  - 5. Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)

1.一组分区(Partition),即数据集基本组成单位;
2.一个计算每个分区的函数;
3.RDD之间依赖关系;
4.一个Partitioner,即RDD分片函数;
5.一个列表,存储存取每个Partition的优先位置(preferred location)

1.3.1.3 RDD 特点

RDD表示只读分区数据集,对RDD进行改动,只能通过RDD转换操作,由一个RDD得到一个新的RDD,新的RDD包含了从其他RDD衍生所必需的信息,RDDs之间存在依赖,RDD执行是按照血缘关系延时计算,如果血缘关系较长,可以通过持久化RDD来切断血缘关系.

1.3.1.3.1 弹性
1.3.1.3.2 分区

RDD逻辑上是分区的,每个分区数据是抽象存在的,计算时会通过一个compute函数得到每个分区数据,如果RDD是通过已有文件系统构建,则compute函数是读取指定文件系统中数据,如果RDD是通过其他RDD转换而来,则compute函数是执行转换逻辑将其他RDD数据进行转换.

1.3.1.3.3 只读

RDD是只读的,要想改变RDD中数据,只能在现有RDD基础上创建新的RDD.

由一个RDD转换到另一个RDD,可以通过丰富的操作算子实现,不再像MapReduce那样只能写map和reduce.

RDD操作算子包括两类,一类是transformations,它是用来将RDD进行转化,构建RDD的血缘关系,另一类是actions,它是用来触发RDD计算得到RDD相关计算结果或者将RDD保存文件系统中.

1.3.1.3.4 依赖

enter image description here

如图所示,RDDs通过操作算子进行转换,转换得到新RDD包含了从其他RDDs衍生所必需的信息,RDDs之间维护着这种血缘关系,也称之为依赖.

依赖包括两种,一种是窄依赖,RDDs之间分区是一一对应,另一种是宽依赖,下游RDD的每个分区与上游RDD(也称之为父RDD)的每个分区都有关,是多对多关系.

1.3.1.3.5 缓存

enter image description here

如果在应用程序中多次使用同一个RDD时,可以将该RDD缓存起来,该RDD只有在第一次计算时会根据血缘关系得到分区数据,在后续其他地方用到该RDD时,会直接从缓存处取而不用再根据血缘关系计算,这样就加速后期的重用.

如图所示,RDD-1经过一系列转换后得到RDD-n并保存到HDFS,RDD-1在这一过程中会有个中间结果,如果将其缓存到内存,那么在随后RDD-1转换到RDD-m这一过程中,就不会计算其之前的RDD-0.

1.3.1.3.6 CheckPoint

虽然RDD血缘关系天然地可以实现容错,当RDD某个分区数据失败或丢失,可以通过血缘关系重建,但是对于长时间迭代型应用来说随着迭代进行,RDDs之间血缘关系会越来越长,一旦在后续迭代过程中出错,则需要通过非常长的血缘关系去重建,势必影响性能.

为此,RDD支持checkpoint将数据保存到持久化存储中,这样就可以切断之前血缘关系,因为checkpoint后的RDD不需要知道它的父RDDs,它可以从checkpoint处拿到数据.

1.3.2 RDD 编程

1.3.2.1 编程模型

在Spark中,RDD被表示为对象,通过对象方法调用RDD进行转换,经过一系列的transformations定义RDD之后,就可以调用actions触发RDD计算,action可以是向应用程序返回结果(count,collect等),或者是向存储系统保存数据(saveAsTextFile等).
在Spark中,只有遇到action才会执行RDD计算(即延迟计算),这样在运行时可以通过管道方式传输多个转换.
使用Spark开发者需要编写一个Driver程序,它被提交到集群以调度运行Worker,Driver中定义了一个或多个RDD.并调用RDD上的action.Worker则执行RDD分区计算任务.

1.3.2.2 RDD 创建
1.3.2.1 集合创建RDD
scala> val rdd = sc.parallelize(Array(511,611,711))

rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> rdd.collect

res0: Array[Int] = Array(511, 611, 711)

scala> 

scala> val makerdd = sc.makeRDD(Array(511,611,711))

makerdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at makeRDD at <console>:24

scala> makerdd.collect

res1: Array[Int] = Array(511, 611, 711)

scala>
1.3.2.2 外部存储系统数据集创建RDD
scala> sc.textFile("/opt/module/spark/input/wordcount/wordcount_001.txt")

res2: org.apache.spark.rdd.RDD[String] = /opt/module/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[3] at textFile at <console>:25

scala>
1.3.2.3 从其他创建RDD
1.3.2.3 RDD 转换
1.3.2.3.1 Value 类型
1.3.2.3.1.1 map(func) Method
scala> val rdd = sc.parallelize(Array(511,611,711))

rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24

scala> rdd.collect

res0: Array[Int] = Array(511, 611, 711)

scala> 

scala> rdd.map((_,1)).collect

res4: Array[(Int, Int)] = Array((511,1), (611,1), (711,1))

scala> 

scala> rdd.map((_*2)).collect

res5: Array[Int] = Array(1022, 1222, 1422)

scala>
1.3.2.3.1.2 mapPartitions(func) Method
scala> rdd.mapPartitions(_.map(_*2)).collect

res11: Array[Int] = Array(1022, 1222, 1422)

scala>
1.3.2.3.1.3 mapPartitionsWithIndex(func) Method
scala> rdd.mapPartitionsWithIndex((index,items)=>(items.map((index,_)))).collect

res13: Array[(Int, Int)] = Array((1,511), (2,611), (3,711))

scala>
1.3.2.3.1.4 flatMap(func) Method
scala> val text = sc.textFile("/core_flow/spark/input/wordcount/wordcount_001.txt")

text: org.apache.spark.rdd.RDD[String] = /core_flow/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[15] at textFile at <console>:24

scala> text.flatMap(_.split(" ")).collect

res16: Array[String] = Array(hadoop, spark, hive, hadoop, spark, hadoop, hbase, flume, hive, scala, java, oozie)

scala>
1.3.2.3.1.5 map()与mapPartition()区别
1.3.2.3.1.6 glom Method
scala> rdd.glom.collect

res17: Array[Array[Int]] = Array(Array(), Array(511), Array(611), Array(711))   

scala>
1.3.2.3.1.7 groupBy(func) Method
scala> rdd.groupBy(_ % 2).collect

res18: Array[(Int, Iterable[Int])] = Array((1,CompactBuffer(611, 711, 511)))    

scala>
1.3.2.3.1.8 filter(func) Method
scala> rdd.filter(_%3==0).collect

res20: Array[Int] = Array(711)

scala>
1.3.2.3.1.9 sample(withReplacement,fraction,seed) Method
scala> val rdd = sc.parallelize(1 to 100)

rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[22] at parallelize at <console>:24

scala> rdd.sample(false,0.1,3).collect

res22: Array[Int] = Array(1, 33, 37, 50, 59, 69, 75, 78, 85, 98) 

scala>
1.3.2.3.1.10 distinct([numTasks])) Method
scala> rdd.distinct(4).collect

res23: Array[Int] = Array(84, 100, 96, 52, 56, 4, 76, 16, 28, 80, 48, 32, 36, 24, 64, 92, 40, 72, 8, 12, 20, 60, 44, 88, 68, 13, 41, 61, 81, 21, 77, 53, 97, 25, 29, 65, 73, 57, 93, 33, 37, 45, 1, 89, 17, 69, 9, 85, 49, 5, 34, 82, 66, 22, 54, 98, 46, 30, 14, 50, 62, 42, 74, 90, 6, 70, 18, 38, 86, 58, 78, 26, 94, 10, 2, 19, 39, 15, 47, 71, 55, 95, 79, 59, 11, 35, 27, 75, 51, 23, 63, 83, 67, 3, 7, 91, 31, 87, 43, 99)

scala>
1.3.2.3.1.11 coalesce(numPartitions) Method
scala> val rdd = sc.parallelize(1 to 16,4)

rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[27] at parallelize at <console>:24



scala> rdd.partitions.size

res24: Int = 4



scala> val coalesceRDD = rdd.coalesce(3)

coalesceRDD: org.apache.spark.rdd.RDD[Int] = CoalescedRDD[28] at coalesce at <console>:26



scala> coalesceRDD.partitions.size

res25: Int = 3

scala>
1.3.2.3.1.12 repartition(numPartitions) Method
scala> val rdd = sc.parallelize(1 to 16,4)

rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[29] at parallelize at <console>:24



scala> rdd.partitions.size

res26: Int = 4



scala> val rerdd = rdd.repartition(2)

rerdd: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[33] at repartition at <console>:26



scala> rerdd.partitions.size

res27: Int = 2

scala>
1.3.2.3.1.13 coalesce与repartition区别

1.coalesce重新分区,可以选择是否进行shuffle过程,由参数shuffle: Boolean = false/true决定.

2.repartition实际上是调用coalesce,进行shuffle过程,源码演示:

def repartition(numpartitions: int)(implicit ord: ordering[t] = null): rdd[t] = withscope {

coalesce(numpartitions, shuffle = true)

}
1.3.2.3.1.14 sortBy(func,[ascending],[numTasks]) Method
scala>  val rdd = sc.parallelize(List(2,1,3,4))

rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[34] at parallelize at <console>:24



scala> rdd.sortBy(x => x).collect()

res29: Array[Int] = Array(1, 2, 3, 4)



scala> rdd.sortBy(x => x%3).collect()

res30: Array[Int] = Array(3, 1, 4, 2)



scala> rdd.sortBy(x => x,false).collect()

res31: Array[Int] = Array(4, 3, 2, 1)



scala>
1.3.2.3.1.15 pipe(command,[envVars]) Method
[root@systemhub511 ~]# vim /opt/module/spark/input/pipe.sh

[root@systemhub511 ~]# chmod 777 /opt/module/spark/input/pipe.sh

#!/bin/

shecho"Start"

while read LINE;do

    echo ">>>" ${LINE}

done

scala> rdd.pipe("/opt/module/spark/pipe.sh").collect

res18: Array[String] = Array(Start, >>>hi, >>>Hello, >>>how, >>>are, >>>you)

scala>
1.3.2.3.2 双Value类型交互
1.3.2.3.2.1 union(otherDataset) Method
scala> var rdd1 = sc.parallelize(1 to 5)

rdd1: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[0] at parallelize at <console>:24



scala> var rdd2 = sc.parallelize(5 to 10)

rdd2: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[1] at parallelize at <console>:24



scala> rdd1.union(rdd2).collect

res0: Array[Int] = Array(1, 2, 3, 4, 5, 5, 6, 7, 8, 9, 10)

scala>
1.3.2.3.2.2 subtract(otherDataset) Method
scala> rdd1.subtract(rdd2).collect

res0: Array[Int] = Array(2, 4, 1, 3)

scala>
1.3.2.3.2.3 intersection(otherDataset) Method
scala> rdd1.intersection(rdd2).collect

res1: Array[Int] = Array(5)

scala>
1.3.2.3.2.4 cartesian(otherDataset) Method
scala> rdd1.cartesian(rdd2).collect

res2: Array[(Int, Int)] = Array((1,5), (1,6), (1,7), (2,5), (2,6), (2,7), (1,8), (1,9), (1,10), (2,8), (2,9), (2,10), (3,5), (3,6), (3,7), (4,5), (4,6), (4,7), (5,5), (5,6), (5,7), (3,8), (3,9), (3,10), (4,8), (4,9), (4,10), (5,8), (5,9), (5,10))

scala>
1.3.2.3.2.5 zip(otherDataset) Method
scala> rdd1.zip(rdd2).collect

res4: Array[(Int, Int)] = Array((1,6), (2,7), (3,8), (4,9), (5,10))

scala>
1.3.2.3.3 Key-Value 类型
1.3.2.3.3.1 partitionBy Method
scala> val rdd1 = sc.parallelize(Array((1,"A"),(2,"B"),(3,"C"),(4,"D")),4)

rdd1: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[0] at parallelize at <console>:24



scala> rdd1.mapPartitionsWithIndex((i,t)=>t.map((i,_))).collect

res3: Array[(Int, (Int, String))] = Array((0,(1,A)), (1,(2,B)), (2,(3,C)), (3,(4,D)))



scala> rdd1.partitionBy(new org.apache.spark.HashPartitioner(2))

res5: org.apache.spark.rdd.RDD[(Int, String)] = ShuffledRDD[3] at partitionBy at <console>:27



scala> res5.partitions.size

res6: Int = 2

scala>
1.3.2.3.3.2 reduceByKey(func,[numTasks]) Method
scala> val rdd = sc.parallelize(List(("female",1),("male",5),("female",5),("male",2)))

rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[4] at parallelize at <console>:24



scala> rdd.reduceByKey((x,y)=>x+y).collect

res7: Array[(String, Int)] = Array((female,6), (male,7))

scala>
1.3.2.3.3.3 groupByKey Method
scala> rdd.groupByKey(2).collect

res8: Array[(String, Iterable[Int])] = Array((female,CompactBuffer(5, 1)), (male,CompactBuffer(5, 2)))



scala>
1.3.2.3.3.4 reduceByKey与groupByKey 区别

1.reduceByKey : 按照key进行聚合,在shuffle之前有combine(预聚合)操作,返回结果是RDD[k,v]

2.groupByKey : 按照key进行分组,直接进行shuffle

3.开发指导 : reduceByKey比groupByKey,建议使用reduceByKey,但是需要注意是否会影响业务逻辑.

1.3.2.3.3.5 aggregateByKey Method

参数 : (zeroValue:U,[partitioner:Partitioner])(seqOp: (U, V) => U,combOp: (U, U) => U)

1.作用 : 在kv对的RDD中,按key将value进行分组合并,合并时将每个value和初始值作为seq函数参数进行计算,返回结果作为一个新的kv对,然后再将结果按照key进行合并,最后将每个分组的value传递给combine函数进行计算(先将前两个value进行计算,将返回结果和下一个value传给combine函数,以此类推),将key与计算结果作为一个新的kv对输出.

2.参数描述 :
zeroValue : 给每一个分区中的每一个key一个初始值.
seqOp : 函数用于在每一个分区中用初始值逐步迭代value
combOp : 函数用于合并每个分区中的结果

enter image description here

scala> val rdd = sc.parallelize(List(("a",3),("a",2),("c",4),("b",3),("c",6),("c",8)),2)

rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[7] at parallelize at <console>:24



scala> rdd.aggregateByKey(0)(math.max(_,_),_+_).collect

res9: Array[(String, Int)] = Array((b,3), (a,3), (c,12))

scala> 

scala> rdd.aggregateByKey(0)(_+_,_+_).collect

res10: Array[(String, Int)] = Array((b,3), (a,5), (c,18))



scala> rdd.reduceByKey(_+_).collect

res11: Array[(String, Int)] = Array((b,3), (a,5), (c,18))

scala>
1.3.2.3.3.6 foldByKey Method
scala> rdd.foldByKey(0)(_+_).collect

res12: Array[(String, Int)] = Array((b,3), (a,5), (c,18))

scala>
1.3.2.3.3.7 combineByKey[C] Method

enter image description here

scala> rdd.combineByKey((_,1),(acc:(Int,Int),v)=>(acc._1+v,acc._2+1),(acc1:(Int,Int),acc2:(Int,Int))=>(acc1._1+acc2._1,acc1._2+acc2._2)).collect

res15: Array[(String, (Int, Int))] = Array((b,(3,1)), (a,(5,2)), (c,(18,3)))    

scala>
1.3.2.3.3.8 sortByKey([ascending],[numTasks]) Method
scala>  rdd.sortByKey().collect

res17: Array[(String, Int)] = Array((a,3), (a,2), (b,3), (c,6), (c,8), (c,4))



scala> rdd.sortByKey(false).collect

res19: Array[(String, Int)] = Array((c,4), (c,6), (c,8), (b,3), (a,3), (a,2))

scala>
1.3.2.3.3.9 mapValues Method
scala> rdd.mapValues(_*2).collect

res20: Array[(String, Int)] = Array((a,6), (a,4), (c,8), (b,6), (c,12), (c,16))

scala>
1.3.2.3.3.10 join(otherDataset,[numTasks]) Method
scala> val rdd = sc.parallelize(Array((1,"a"),(2,"b"),(3,"c")))

rdd: org.apache.spark.rdd.RDD[(Int, String)] = ParallelCollectionRDD[20] at parallelize at <console>:24



scala> val rdd1 = sc.parallelize(Array((1,4),(2,5),(3,6)))

rdd1: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[21] at parallelize at <console>:24



scala> rdd.join(rdd1).collect

res21: Array[(Int, (String, Int))] = Array((1,(a,4)), (2,(b,5)), (3,(c,6))) 



scala> rdd.leftOuterJoin(rdd1).collect

res22: Array[(Int, (String, Option[Int]))] = Array((1,(a,Some(4))), (2,(b,Some(5))), (3,(c,Some(6))))



scala> rdd.rightOuterJoin(rdd1).collect

res23: Array[(Int, (Option[String], Int))] = Array((1,(Some(a),4)), (2,(Some(b),5)), (3,(Some(c),6)))

scala>
1.3.2.3.3.11 cogroup(otherDataset,[numTasks]) Method
scala> rdd.cogroup(rdd1).collect

res24: Array[(Int, (Iterable[String], Iterable[Int]))] = Array((1,(CompactBuffer(a),CompactBuffer(4))), (2,(CompactBuffer(b),CompactBuffer(5))), (3,(CompactBuffer(c),CompactBuffer(6))))



scala>
1.3.2.4 Action
1.3.2.4.1 reduce(func) Method
scala> val rdd = sc.parallelize(1 to 10)

rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[33] at parallelize at <console>:24



scala> rdd.reduce(_+_)

res25: Int = 55

scala>
1.3.2.4.2 collect() Method
scala> rdd.collect

res26: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

scala>
1.3.2.4.3 count() Method
scala> rdd.count

res27: Long = 10

scala>
1.3.2.4.4 first() Method
scala> rdd.first

res28: Int = 1

scala>
1.3.2.4.5 take(n) Method
scala> rdd.take(2)

res30: Array[Int] = Array(1, 2)

scala>
1.3.2.4.6 takeOrdered(n) Method
scala> rdd.takeOrdered(3)

res31: Array[Int] = Array(1, 2, 3) 

scala>
1.3.2.4.7 aggregate Method

参数 : (zeroValue: U)(seqOp: (U, T) ⇒U, combOp: (U, U) ⇒U)

作用 : aggregate函数将每个分区里面的元素通过seqOp和初始值进行聚合,然后用combine函数将每个分区的结果和初始值(zeroValue)进行combine操作,这个函数最终返回的类型不需要和RDD中元素类型一致.

scala> rdd.aggregate(0)(_+_,_+_)

res32: Int = 55

scala>
1.3.2.4.8 fold(num)(func) Method
scala> rdd.fold(0)(_+_)

res34: Int = 55

scala>
1.3.2.4.9 saveAsTextFile(path) Method
1.3.2.4.10 saveAsSequenceFile(path) Method
1.3.2.4.11 saveAsObjectFile(path) Method
1.3.2.4.12 countByKey() Method
scala> val rdd = sc.parallelize(List((1,3),(1,2),(1,4),(2,3),(3,6),(3,8)),3)

rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[35] at parallelize at <console>:24



scala> rdd.countByKey

res35: scala.collection.Map[Int,Long] = Map(3 -> 2, 1 -> 3, 2 -> 1)

scala>
1.3.2.4.13 foreach(func) Method
scala> rdd.foreach(print)
1.3.2.5 RDD 函数传递

在实际开发中往往需要开发者定义一些对于RDD操作,那么此时需要主要的是,初始化工作是在Driver端进行,而实际运行程序是在Executor端进行,这就涉及到了跨进程通信,跨进程通信是需要序列化操作.

1.3.2.5.1 传递方法

在这个方法中所调用的方法isMatch()是定义在Search这个类中,实际上调用的是this.isMatch(),this表示Search这个类的对象,程序在运行过程中需要将Search对象序列化以后传递到Executor端.

1.3.2.5.2 传递属性

在这个方法中所调用的方法query是定义在Search这个类中的字段,实际上调用的是this.query,this表示Search这个类的对象,程序在运行过程中需要将Search对象序列化以后传递到Executor端.

package com.geekparkhub.core.spark.application.methods

import org.apache.spark.rdd.RDD



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * Search

  * <p>

  */



class Search(query: String) extends Serializable {



  // 过滤出包含字符串数据

  def isMatch(s: String): Boolean = {

    s.contains(query)

  }



  // 过滤出包含字符串RDD

  def getMatch1(rdd: RDD[String]): RDD[String] = {

    rdd.filter(isMatch)

  }



  // 过滤出包含字符串RDD

  def getMatche2(rdd: RDD[String]): RDD[String] = {

    rdd.filter(x => x.contains(query))

  }

}

package com.geekparkhub.core.spark.application.methods

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * TransFormAction

  * <p>

  */



object TransFormAction {

  def main(args: Array[String]): Unit = {



    // 创建SpakConf

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("TransFormAction")



    // 创建SC

    val sc = new SparkContext(sparkConf)



    // 创建RDD

    val word: RDD[String] = sc.parallelize(Array("abc", "dcd"))



    // 创建Search对象

    val search = new Search("a")



    // 调用方法

    val searched: RDD[String] = search.getMatch1(word)



    // 循环输出

    searched.collect().foreach(println)



    // 关闭资源

    sc.stop()

  }

}
1.3.2.6 RDD 依赖关系
1.3.2.6.1 Lineage

enter image description here

RDD只支持粗粒度转换,即在大量记录上执行的单个操作,将创建RDD的一系列Lineage(血统)记录下来,以便恢复丢失的分区,RDD的Lineage会记录RDD的元数据信息和转换行为,当该RDD的部分分区数据丢失时,它可以根据这些信息来重新运算和恢复丢失的数据分区.

scala> sc.textFile("/core_flow/spark/input/wordcount/wordcount_001.txt")

res0: org.apache.spark.rdd.RDD[String] = /core_flow/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[1] at textFile at <console>:25



scala> res0.flatMap(_.split(" "))

res2: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[2] at flatMap at <console>:27



scala> res2.map((_,1))

res3: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[3] at map at <console>:29



scala> res3.reduceByKey(_+_)

res4: org.apache.spark.rdd.RDD[(String, Int)] = ShuffledRDD[4] at reduceByKey at <console>:31



scala>

scala> res0.toDebugString

res5: String =

(2) /core_flow/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[1] at textFile at <console>:25 []

 |  /core_flow/spark/input/wordcount/wordcount_001.txt HadoopRDD[0] at textFile at <console>:25 []

scala> 

scala> res2.toDebugString

res6: String =

(2) MapPartitionsRDD[2] at flatMap at <console>:27 []

 |  /core_flow/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[1] at textFile at <console>:25 []

 |  /core_flow/spark/input/wordcount/wordcount_001.txt HadoopRDD[0] at textFile at <console>:25 []

scala> 

scala> res3.toDebugString

res7: String =

(2) MapPartitionsRDD[3] at map at <console>:29 []

 |  MapPartitionsRDD[2] at flatMap at <console>:27 []

 |  /core_flow/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[1] at textFile at <console>:25 []

 |  /core_flow/spark/input/wordcount/wordcount_001.txt HadoopRDD[0] at textFile at <console>:25 []

scala> 

scala> res4.toDebugString

res8: String =

(2) ShuffledRDD[4] at reduceByKey at <console>:31 []

 +-(2) MapPartitionsRDD[3] at map at <console>:29 []

    |  MapPartitionsRDD[2] at flatMap at <console>:27 []

    |  /core_flow/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[1] at textFile at <console>:25 []

    |  /core_flow/spark/input/wordcount/wordcount_001.txt HadoopRDD[0] at textFile at <console>:25 []

scala>
1.3.2.6.2 窄依赖

enter image description here

1.3.2.6.3 宽依赖

enter image description here

1.3.2.6.4 DAG

enter image description here

1.3.2.6.5 任务划分(重点)
1.3.2.6.7 RDD缓存
1.3.2.6.8 RDDCheckPoint
scala> sc.setCheckpointDir("hdfs://systemhub511:9000/core_flow/spark/checkpoint")

scala>  val rdd = sc.parallelize(Array("systemhub511"))

rdd: org.apache.spark.rdd.RDD[String] = ParallelCollectionRDD[5] at parallelize at <console>:24

scala> val check = rdd.map(_+System.currentTimeMillis)

check: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[6] at map at <console>:26

scala> 

scala> check.collect

res10: Array[String] = Array(systemhub5111559138263898)



scala> check.collect

res11: Array[String] = Array(systemhub5111559138266443)



scala> check.collect

res12: Array[String] = Array(systemhub5111559138267862)



scala>

1.3.3 Key-Value RDD 数据分区

1.3.3.1 获取RDD 分区
scala> rdd.partitioner

res14: Option[org.apache.spark.Partitioner] = None
1.3.3.2 Hash 分区
scala> val nopar = sc.parallelize(List((1,3),(1,2),(2,4),(2,3),(3,6),(3,8)),8)

nopar: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[7] at parallelize at <console>:25



scala> nopar.mapPartitionsWithIndex((index,iter)=>{Iterator(index.toString+":"+iter.mkString("|"))}).collect

res15: Array[String] = Array(0:, 1:(1,3), 2:(1,2), 3:(2,4), 4:, 5:(2,3), 6:(3,6), 7:(3,8))



scala> val hashpar = nopar.partitionBy(new org.apache.spark.HashPartitioner(7))

hashpar: org.apache.spark.rdd.RDD[(Int, Int)] = ShuffledRDD[9] at partitionBy at <console>:27



scala> hashpar.count

res20: Long = 6



scala> hashpar.partitioner

res21: Option[org.apache.spark.Partitioner] = Some(org.apache.spark.HashPartitioner@7)



scala> hashpar.mapPartitions(iter => Iterator(iter.length)).collect()

res22: Array[Int] = Array(0, 2, 2, 2, 0, 0, 0)

scala>
1.3.3.3 Ranger 分区
1.3.3.4 自定义 分区
package com.geekparkhub.core.spark.application.partitioner



import org.apache.spark.Partitioner



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * CustomerPartitioner

  * <p>

  */



class CustomerPartitioner(partitions: Int) extends Partitioner {

  override def numPartitions: Int = partitions

  override def getPartition(key: Any): Int = {

    0

  }

}

package com.geekparkhub.core.spark.application.partitioner



import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * PartitionerAction

  * <p>

  */



object PartitionerAction {



  def main(args: Array[String]): Unit = {

    // 创建SpakConf

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("TransFormAction")



    // 创建SC

    val sc = new SparkContext(sparkConf)



    // 创建RDD

    val word: RDD[String] = sc.parallelize(Array("abc", "dcd"))



    // 将元素转换为元祖

    val wordAndOne: RDD[(String, Int)] = word.map((_, 1))



    // 自定义分区

    val partitioned: RDD[(String, Int)] = wordAndOne.partitionBy(new CustomerPartitioner(2))



    // 查看分区后分区结果

    val indexAndData: RDD[(Int, (String, Int))] = partitioned.mapPartitionsWithIndex((i,t)=>t.map((i,_)))



    // 打印数据

    indexAndData.collect().foreach(println)



    // 关闭资源

    sc.stop()

  }

}

(0,(abc,1))

(0,(dcd,1))

1.3.4 数据读取&保存

1.3.4.1 文件类数据读取&保存
1.3.4.1 Text File
scala> sc.textFile("hdfs://systemhub511:9000/core_flow/spark/input/wordcount/wordcount_001.txt")

res23: org.apache.spark.rdd.RDD[String] = hdfs://systemhub511:9000/core_flow/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[12] at textFile at <console>:26



scala> res23.toDebugString

res25: String =

(2) hdfs://systemhub511:9000/core_flow/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[12] at textFile at <console>:26 []

 |  hdfs://systemhub511:9000/core_flow/spark/input/wordcount/wordcount_001.txt HadoopRDD[11] at textFile at <console>:26 []

scala> 

scala> hdfsFile.saveAsTextFile("/core_flow/spark/output/wordcount/")
1.3.4.2 Json File
scala> import scala.util.parsing.json.JSON

import scala.util.parsing.json.JSON

scala> 

[root@systemhub511 ~]# hadoop fs -mkdir -p /core_flow/spark/json/001

[root@systemhub511 ~]# hadoop fs -put /opt/module/spark/examples/src/main/resources/people.json /core_flow/spark/json/001/

scala> val json = sc.textFile("hdfs://systemhub511:9000/core_flow/spark/json/001/people.json")

json: org.apache.spark.rdd.RDD[String] = hdfs://systemhub511:9000/core_flow/spark/json/001/people.json MapPartitionsRDD[14] at textFile at <console>:26

scala>

scala> val result = json.map(JSON.parseFull)

result: org.apache.spark.rdd.RDD[Option[Any]] = MapPartitionsRDD[15] at map at <console>:28

scala>

scala> result.collect

res26: Array[Option[Any]] = Array(Some(Map(name -> Michael)), Some(Map(name -> Andy, age -> 30.0)), Some(Map(name -> Justin, age -> 19.0)))

scala>
1.3.4.3 Sequence File
scala>  val rdd = sc.parallelize(Array((1,2),(3,4),(5,6)))

rdd: org.apache.spark.rdd.RDD[(Int, Int)] = ParallelCollectionRDD[16] at parallelize at <console>:26

scala> 

scala> rdd.saveAsSequenceFile("file:///opt/module/spark/seqFile")

[root@systemhub511 ~]# cd /opt/module/spark/seqFile/

[root@systemhub511 seqFile]# ll -a

总用量 28

drwxr-xr-x.  2 root          root          4096 5月  29 23:57 .

drwxr-xr-x. 21 geekdeveloper geekdeveloper 4096 5月  30 00:05 ..

-rw-r--r--.  1 root          root            92 5月  29 23:57 part-00000

-rw-r--r--.  1 root          root            12 5月  29 23:57 .part-00000.crc

-rw-r--r--.  1 root          root           108 5月  29 23:57 part-00003

-rw-r--r--.  1 root          root            12 5月  29 23:57 .part-00003.crc

-rw-r--r--.  1 root          root             0 5月  29 23:57 _SUCCESS

-rw-r--r--.  1 root          root             8 5月  29 23:57 ._SUCCESS.crc

[root@systemhub511 seqFile]# cat part-00000

SEQ org.apache.hadoop.io.IntWritable org.apache.hadoop.io.IntWritabler[-o���]h�~u���

[root@systemhub511 seqFile]#

scala>  val seq = sc.sequenceFile[Int,Int]("file:///opt/module/spark/seqFile")

seq: org.apache.spark.rdd.RDD[(Int, Int)] = MapPartitionsRDD[19] at sequenceFile at <console>:26

scala>

scala> seq.collect

res14: Array[(Int, Int)] = Array((1,2), (3,4), (5,6))
1.3.4.4 ObjectFile
scala> val rdd = sc.parallelize(Array(1,2,3,4))

rdd: org.apache.spark.rdd.RDD[Int] = ParallelCollectionRDD[20] at parallelize at <console>:26

scala> 

scala> rdd.saveAsObjectFile("file:///opt/module/spark/objectFile")

[root@systemhub511 ~]# cd /opt/module/spark/objectFile/

[root@systemhub511 objectFile]# ll

总用量 8

-rw-r--r--. 1 root root 138 5月  30 00:05 part-00000

-rw-r--r--. 1 root root 138 5月  30 00:05 part-00003

-rw-r--r--. 1 root root   0 5月  30 00:05 _SUCCESS

[root@systemhub511 objectFile]# cat part-00000

SEQ!org.apache.hadoop.io.NullWritable"org.apache.hadoop.io.BytesWritable� �L�h�l:T���#��ur[IM�`&v겥xp

[root@systemhub511 objectFile]#

scala> val objFile = sc.objectFile[(Int)]("file:///opt/module/spark/objectFile")

objFile: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[24] at objectFile at <console>:26

scala>

objFile.collect

res19: Array[Int] = Array(1, 2, 3, 4)
1.3.4.2 文件系统数据读取&保存
1.3.4.1 HDFS

Spark整个生态系统与Hadoop是完全兼容,所以对于Hadoop所支持的文件类型或者数据库类型,Spark也同样支持.
另外由于Hadoop的API有新旧两个版本,所以Spark为了能够兼容Hadoop所有版本,也提供了两套创建操作接口.
对于外部存储创建操作而言,hadoopRDD和newHadoopRDD是最为抽象的两个函数接口,主要包含以下四个参数 :

1.输入格式(InputFormat) : 制定数据输入类型,如TextInputFormat等,新旧两个版本所引用版本分别是org.apache.hadoop.mapred.InputFormat和org.apache.hadoop.mapreduce.InputFormat(NewInputFormat)
2.键类型 : 指定[K,V]键值对中K类型
3.值类型: 指定[K,V]键值对中V类型
4.分区值 : 指定由外部存储生成RDD的partition数量最小值,如果没有指定系统会使用默认值defaultMinSplits.

其他创建操作API接口都是为了方便最终Spark程序开发者而设置的,是这两个接口高效实现版本,例如对于textFile而言,只有path这个指定文件路径参数,其他参数在系统内部指定了默认值.
1.在Hadoop中以压缩形式存储数据,不需要指定解压方式就能够进行读取,因为Hadoop本身有一个解压器会根据压缩文件后缀推断解压算法进行解压.
2.如果用Spark从Hadoop中读取某种类型数据不知道怎么读取的时候,上网查找一个使用map-reduce时候是怎么读取这种这种数据,然后再将对应的读取方式改写成上面的hadoopRDD和newAPIHadoopRDD两个类即可.

1.3.4.2 MySQL数据库 连接
<dependencies>

 <dependency>

  <groupId>mysql</groupId>

  <artifactId>mysql-connector-java</artifactId>

  <version>8.0.15</version>

 </dependency>

</dependencies>

package com.geekparkhub.core.spark.application.dataconnections



import java.sql.DriverManager



import org.apache.spark.deploy.worker.DriverWrapper

import org.apache.spark.rdd.JdbcRDD

import org.apache.spark.{SparkConf, SparkContext}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * JDBCConnection

  * <p>

  */



object JDBCConnection {



  def main(args: Array[String]): Unit = {



    // 创建SpakConf

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("JDBCConnection")



    // 创建SC

    val sc = new SparkContext(sparkConf)



    // 定义JDBC连接属性信息

    val driver = "com.mysql.jdbc.Driver"

    val url = "jdbc:mysql://systemhub711:3306/company"

    val userName = "root"

    val passWd = "ax04854"



    // 创建JDBC RDD

    val JdbcRDD = new JdbcRDD[(Int, String)](sc, () => {

      Class.forName(driver)

      DriverManager.getConnection(url, userName, passWd)

    }, "select id,name from staff where ? <= id and id <= ?",

      1,

      10,

      1,

      x => {

        (x.getInt(1), x.getString(2))

      }

    )



    // 打印JdbcRDD结果

    JdbcRDD.collect().foreach(println)



    // 关闭资源

    sc.stop()

  }

}

package com.geekparkhub.core.spark.application.dataconnections



import org.apache.spark.{SparkConf, SparkContext}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * JBDCinsertData

  * <p>

  */



object JBDCinsertData {

  def main(args: Array[String]): Unit = {



    // 创建SpakConf

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("JBDCRead")



    // 创建SC

    val sc = new SparkContext(sparkConf)



    // 创建数据

    val data = sc.parallelize(List("Female", "Male", "Female"))



    // 调用添加数据方法

    data.foreachPartition(insertData)

  }



  // 添加数据方法

  def insertData(iterator: Iterator[String]): Unit = {

    Class.forName("com.mysql.jdbc.Driver").newInstance()

    val conn = java.sql.DriverManager.getConnection("jdbc:mysql://systemhub711:3306/company", "root", "000000")

    iterator.foreach(data => {

      val ps = conn.prepareStatement("insert into staff(name) values(?)")

      ps.setString(1, data)

      ps.executeUpdate()

    })

  }

}
1.3.4.3 HBase 数据库
<dependency>

  <groupId>org.apache.hbase</groupId>

  <artifactId>hbase-server</artifactId>

  <version>1.3.1</version>

</dependency>

<dependency>

  <groupId>org.apache.hbase</groupId>

  <artifactId>hbase-client</artifactId>

  <version>1.3.1</version>

</dependency>

package com.geekparkhub.core.spark.application.dataconnections



import org.apache.hadoop.conf.Configuration

import org.apache.hadoop.hbase.HBaseConfiguration

import org.apache.hadoop.hbase.client.Result

import org.apache.hadoop.hbase.io.ImmutableBytesWritable

import org.apache.hadoop.hbase.mapreduce.TableInputFormat

import org.apache.hadoop.hbase.util.Bytes

import org.apache.spark.rdd.{NewHadoopRDD, RDD}

import org.apache.spark.{SparkConf, SparkContext}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * HbaseConnection

  * <p>

  */



object HbaseConnection {

  def main(args: Array[String]): Unit = {



    // 创建SpakConf

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HbaseConnection")



    // 创建SC

    val sc = new SparkContext(sparkConf)



    //构建HBase配置信息

    val conf: Configuration = HBaseConfiguration.create()

    conf.set("hbase.zookeeper.quorum", "systemhub511,systemhub611,systemhub711")

    conf.set(TableInputFormat.INPUT_TABLE, "test")



    // 读取HBASE数据

    val hbaseRDD = new NewHadoopRDD(sc, classOf[TableInputFormat], classOf[ImmutableBytesWritable], classOf[Result], conf)



    // 获取RowKey

    val value: RDD[String] = hbaseRDD.map(x => Bytes.toString(x._2.getRow))



    // 输出数据

    value.collect().foreach(println)



    // 关闭资源

    sc.stop()

  }

}

package com.geekparkhub.core.spark.application.dataconnections



import org.apache.hadoop.hbase.client.Put

import org.apache.hadoop.hbase.io.ImmutableBytesWritable

import org.apache.hadoop.hbase.mapred.TableOutputFormat

import org.apache.hadoop.hbase.util.Bytes

import org.apache.hadoop.mapred.JobConf

import org.apache.spark.rdd.RDD

import org.apache.spark.{SparkConf, SparkContext}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * HbaseWrite

  * <p>

  */



object HbaseWrite {

  def main(args: Array[String]): Unit = {



    // 创建SpakConf

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HbaseWrite")



    // 创建SC

    val sc = new SparkContext(sparkConf)



    // 创建RDD

    val initialRDD: RDD[(Int, String, Int)] = sc.parallelize(List((1, "apple", 11), (2, "banana", 12), (3, "pear", 13)))



    // 创建JobConf

    val conf = new JobConf()

    conf.set("hbase.zookeeper.quorum", "systemhub511,systemhub611,systemhub711")

    conf.setOutputFormat(classOf[TableOutputFormat[ImmutableBytesWritable]])

    conf.set(TableOutputFormat.OUTPUT_TABLE, "test")



    // 定义 Hbase 添加数据方法

    def convert(triple: (Int, String, Int)): (ImmutableBytesWritable, Put) = {

      val put = new Put(Bytes.toBytes(triple._1))

      put.addImmutable(Bytes.toBytes("info"), Bytes.toBytes("name"), Bytes.toBytes(triple._2))

      put.addImmutable(Bytes.toBytes("info"), Bytes.toBytes("price"), Bytes.toBytes(triple._3))(new ImmutableBytesWritable, put)

    }



    // 转换RDD

    val writRDD: RDD[(ImmutableBytesWritable, Put)] = initialRDD.map(convert)



    // 写入HBASE

    writRDD.saveAsHadoopDataset(conf)



    // 关闭资源

    sc.stop()

  }

}

1.3.5 RDD 编程进阶

1.3.5.1 累加器

累加器用来对信息进行聚合,通常在向Spark传递函数时,比如使用map()函数或者用filter()传条件时,可以使用驱动器程序中定义变量,但是集群中运行每个任务都会得到这些变量的一份新副本,更新这些副本的值也不会影响驱动器中的对应变量,如果想实现所有分片处理时更新共享变量的功能,那么累加器可以实现想要的效果.

1.3.5.1.1 系统累加器

通过在驱动器中调用SparkContext.accumulator(initialValue)方法,创建出存有初始值的累加器,返回值为org.apache.spark.Accumulator[T]对象,其中T是初始值initialValue的类型,Spark闭包里的执行器代码可以使用累加器的+=方法(在Java中是add)增加累加器的值,驱动器程序可以调用累加器的value属性(在Java中使用value()或setValue())来访问累加器的值.

工作节点上任务不能访问累加器值,从这些任务的角度来看,累加器是一个只写变量.

对于要在行动操作中使用累加器,Spark只会把每个任务对各累加器的修改应用一次,因此,如果想要一个无论在失败还是重复计算时都绝对可靠的累加器,必须把它放在foreach()这样的行动操作中,转化操作中累加器可能会发生不止一次更新.

package com.geekparkhub.core.spark.application.methods



import org.apache.spark.rdd.RDD

import org.apache.spark.util.LongAccumulator

import org.apache.spark.{SparkConf, SparkContext}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * AccuAction

  * <p>

  */



object AccuAction {

  def main(args: Array[String]): Unit = {



    // 创建SpakConf

    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("AccuAction")



    // 创建SC

    val sc = new SparkContext(sparkConf)



    // 累加器

    val sum: LongAccumulator = sc.longAccumulator("sum")



    // 创建RDD

    val value: RDD[Int] = sc.parallelize(Array(1, 2, 3, 4))



    val word: RDD[(Int, Int)] = value.map(x => {

      // 添加累加

      sum.add(x)

      (x, 1)

    })



    word.collect().foreach(println)



    println(sum.value)



    // 关闭资源

    sc.stop()

  }

}
1.3.5.1.2 自定义累加器

自定义累加器类型功能在1.X版本中就已经提供,但是使用起来比较麻烦,在2.0版本后,累加器的易用性有了较大改进,而且官方还提供了一个新抽象类 : AccumulatorV2来提供更加友好自定义类型累加器的实现方式,实现自定义类型累加器需要继承AccumulatorV2并至少覆写下例中出现的方法,

package com.geekparkhub.core.spark.application.methods



import org.apache.spark.util.AccumulatorV2



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * AccumulatorAction

  * <p>

  */



class AccumulatorAction extends AccumulatorV2[Int,Int]{



  var sum  = 0



  // 判断是否为空

  override def isZero: Boolean = sum == 0



  // 复制方法

  override def copy(): AccumulatorV2[Int, Int] = {

    val accumulatorAction = new AccumulatorAction

    accumulatorAction.sum = this.sum

    accumulatorAction

  }



  // 重置方法

  override def reset(): Unit = 0



  // 累加方法

  override def add(v: Int): Unit = sum += v



  // 合并方法

  override def merge(other: AccumulatorV2[Int, Int]): Unit = sum += other.value



  // 返回值

  override def value: Int = sum

}
1.3.5.2 广播变量 (调优策略)

广播变量用来高效分发较大对象,向所有工作节点发送一个较大的只读值,以供一个或多个Spark操作使用.

比如,如果应用需要向所有节点发送一个较大的只读查询表,甚至是机器学习算法中的一个很大的特征向量,广播变量用起来都很顺手,在多个并行操作中使用同一个变量,但是Spark会为每个任务分别发送.

使用广播变量过程 :
1.通过对一个类型T的对象调用SparkContext.broadcast创建出Broadcast[T]对象,任何可序列化类型都可以这么实现.
2.通过value属性访问该对象值(在Java中为value()方法).
3.变量只会被发到各个节点一次,应作为只读值处理(修改这个值不会影响到别的节点).

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3))

broadcastVar: org.apache.spark.broadcast.Broadcast[Array[Int]] = Broadcast(0)



scala> broadcastVar.value

res0: Array[Int] = Array(1, 2, 3)



scala>

🔥 1.4 Spark SQL 🔥

1.4.1 Spark SQL 概述

1.4.1.1 什么是 Spark SQL

Spark SQL是Spark用来处理结构化数据模块,它提供了2个编程抽象 : DataFrame和DataSet,并且作为分布式SQL查询引擎作用.

已经学习了Hive,它是将Hive SQL转换成MapReduce然后提交到集群上执行,大大简化了编写MapReduc程序复杂性,由于MapReduce计算模型执行效率比较慢,所以Spark SQL应运而生,它是将Spark SQL转换成RDD,然后提交到集群执行,执行效率非常快.

1.4.1.2 Spark SQL 特点
1.4.1.3 什么是 DataFrame

与RDD类似,DataFrame也是一个分布式数据容器,然而DataFrame更像传统数据库的二维表格,除了数据以外,还记录数据的结构信息,即schema,同时与Hive类似,DataFrame也支持嵌套数据类型(struct / array / map).

从API易用性角度上看,DataFrame API提供是一套高层的关系操作,比函数式RDD API要更加友好,门槛更低.

enter image description here

上图直观地体现了DataFrame和RDD区别,左侧RDD[Person]虽然以Person为类型参数,但Spark框架本身不了解Person类内部结构,而右侧DataFrame却提供了详细的结构信息,使得Spark SQL可以清楚地知道该数据集中包含哪些列,每列名称和类型各是什么,DataFrame是为数据提供了Schema视图,可以把它当做数据库中一张数据表.

DataFrame也是懒执行,性能上比RDD要高要原因 : 优化执行计划,查询计划通过Spark catalyst optimiser进行优化.

enter image description here

为了说明查询优化,上图展示的人口数据分析示例,图中构造了两个DataFrame,将它们join之后又做了一次filter操作,如果原封不动地执行这个执行计划,最终的执行效率是不高的,因为join是一个代价较大操作,也可能会产生一个较大数据集,如果能将filter下推到join下方,先对DataFrame进行过滤,再join过滤后的较小的结果集,便可以有效缩短执行时间.
而Spark SQL的查询优化器正是这样做的,简而言之逻辑查询计划优化就是一个利用基于关系代数的等价变换,将高成本的操作替换为低成本操作的过程.

1.4.1.4 什么是 DataSet

1.DataSet是DataframeAPI扩展,是SparkSQL最新数据抽象.

2.友好API风格,既具有类型安全检查也具有Dataframe的查询优化特性.

3.Dataset支持编解码器,当需要访问非堆上的数据时可以避免反序列化整个对象,提高了效率.

4.样例类被用来在Dataset中定义数据结构信息,样例类中每个属性的名称直接映射到DataSet中的字段名称.

5.Dataframe是Dataset的特列,DataFrame=Dataset[Row],所以可以通过as方法将Dataframe转换为Dataset,Row是一个类型,跟Car / Person这些类型一样,所有表结构信息都用Row来表示.

6.DataSet是强类型,比如可以有Dataset[Car],Dataset[Person].

7.DataFrame只是知道字段,但是不知道字段类型,所以在执行这些操作时是没办法在编译的时候检查是否类型失败,比如可以对一个String进行减法操作,在执行时才报错,而DataSet不仅仅知道字段,而且知道字段类型,所以有更严格的错误检查,就跟JSON对象和类对象之间的类比.

1.4.2 Spark SQL 编程

1.4.2.1 SparkSession 新起始点

在老版本中,SparkSQL提供两种SQL查询起始点 :
SQLContext : 用于Spark提供SQL查询.
HiveContext : 用于连接Hive查询.

SparkSession是Spark最新SQL查询起始点,实质上是SQLContext和HiveContext组合,所以在SQLContext和HiveContext上可用API在SparkSession上同样是可以使用,SparkSession内部封装了sparkContext,所以计算实际上是由sparkContext完成.

1.4.2.2 DataFrame
1.4.2.2.1 创建

在SparkSQL中SparkSession是创建DataFrame和执行SQL入口.
创建DataFrame有三种方式 :
1.通过Spark数据源进行创建.
2.从已存在的RDD进行转换.
3.从Hive Table进行查询返回.

scala> spark.read.

csv      jdbc   load     options   parquet   table   textFile      

format   json   option   orc       schema    text 

scala> spark.read.

scala> val jsonflow = spark.read.json("hdfs://systemhub511:9000/core_flow/spark/json/001/people.json")

jsonflow: org.apache.spark.sql.DataFrame = [age: bigint, name: string]=



scala> jsonflow.show

+----+-------+

| age|   name|

+----+-------+

|null|Michael|

|  30|   Andy|

|  19| Justin|

+----+-------+



scala>
1.4.2.2.2 SQL风格语法(主要)
scala> jsonflow.createTempView("people")

scala> val sqlDF = spark.sql("SELECT * FROM people")

sqlDF: org.apache.spark.sql.DataFrame = [age: bigint, name: string]



scala> sqlDF.show

+----+-------+

| age|   name|

+----+-------+

|null|Michael|

|  30|   Andy|

|  19| Justin|

+----+-------+



scala> 

scala> jsonflow.createGlobalTempView("peoples")

scala> spark.sql("SELECT * FROM global_temp.peoples").show()

+----+-------+

| age|   name|

+----+-------+

|null|Michael|

|  30|   Andy|

|  19| Justin|

+----+-------+



scala> spark.newSession().sql("SELECT * FROM global_temp.peoples").show()

+----+-------+

| age|   name|

+----+-------+

|null|Michael|

|  30|   Andy|

|  19| Justin|

+----+-------+



scala>
1.4.2.2.3 DSL风格语法(次要)
scala> jsonflow.printSchema

root

 |-- age: long (nullable = true)

 |-- name: string (nullable = true)



scala> 

scala> jsonflow.select("name").show

+-------+

|   name|

+-------+

|Michael|

|   Andy|

| Justin|

+-------+



scala> 

scala> jsonflow.select($"name",$"age" + 1).show()

+-------+---------+

|   name|(age + 1)|

+-------+---------+

|Michael|     null|

|   Andy|       31|

| Justin|       20|

+-------+---------+



scala>

scala> jsonflow.filter($"age" > 21).show()

+---+----+

|age|name|

+---+----+

| 30|Andy|

+---+----+



scala> 

scala> jsonflow.groupBy("age").count().show()

+----+-----+

| age|count|

+----+-----+

|  19|    1|

|null|    1|

|  30|    1|

+----+-----+



scala>
1.4.2.2.4 RDD转换为DateFrame

如果需要RDD与DF或者DS之间操作,需要引入import spark.implicits._
spark并不是包名,而是sparkSession对象名称.

scala> import spark.implicits._

import spark.implicits._



scala> val peopleRDD = sc.textFile("hdfs://systemhub511:9000/core_flow/spark/input/wordcount/wordcount_001.txt")

peopleRDD: org.apache.spark.rdd.RDD[String] = hdfs://systemhub511:9000/core_flow/spark/input/wordcount/wordcount_001.txt MapPartitionsRDD[30] at textFile at <console>:27



scala>

scala> peopleRDD.map{x=>{val split = x.split(",");(split(0),split(1).trim)}}.toDF("name","age")

res11: org.apache.spark.sql.DataFrame = [name: string, age: string]

scala> 

scala> case class People(name:String, age:Int)

defined class People



scala> peopleRDD.map{x=>{val split = x.split(",");People(split(0),split(1).trim.toInt)}}.toDF

res17: org.apache.spark.sql.DataFrame = [name: string, age: int]



scala> res17.toDF

res18: org.apache.spark.sql.DataFrame = [name: string, age: int]

scala> 

scala> import org.apache.spark.sql.types._

import org.apache.spark.sql.types._

scala> 

scala> val structType: StructType = StructType(StructField("name",StringType) :: StructField("age",IntegerType) :: Nil)

structType: org.apache.spark.sql.types.StructType = StructType(StructField(name,StringType,true), StructField(age,IntegerType,true))

scala> 

scala> import org.apache.spark.sql.Row

import org.apache.spark.sql.Row

scala> 

scala> val data = peopleRDD.map{x => val para = x.split(",");Row(para(0),para(1).trim.toInt)}

scala> 

scala> val dataFrame = spark.createDataFrame(data, structType)

package com.geekparkhub.core.spark.application.sparksql



import org.apache.spark.rdd.RDD

import org.apache.spark.sql.types.{IntegerType, StructField, StructType}

import org.apache.spark.{SparkConf, SparkContext}

import org.apache.spark.sql.{DataFrame, Row, SparkSession}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * SqlAction

  * <p>

  */



object SqlAction {

  def main(args: Array[String]): Unit = {



    // 创建SparkSession

    val sparkSession: SparkSession = SparkSession

      .builder().master("local[*]").appName("SqlAction").getOrCreate()



    // 创建SC

    val sc: SparkContext = sparkSession.sparkContext



    // 创建 RDD

    val rdd: RDD[Int] = sc.parallelize(Array(1,2,3,4,5))



    // 将Int类型RDD转换为Row类型RDD

    val rowRDD: RDD[Row] = rdd.map(x => {Row(x)})



    // 数据输出

    rowRDD.collect().foreach(println)



    // 创建元数据信息

    val structType = new StructType

    val structTypes: StructType = structType.add(StructField("id", IntegerType))

    val dataFrame: DataFrame = sparkSession.createDataFrame(rowRDD,structTypes)



    // 导入隐式转换

    import sparkSession.implicits._



    // DSL风格 数据查询

    dataFrame.select("id").show()



    // 关闭资源

    sparkSession.stop()

  }

}
1.4.2.2.5 DateFrame转换为RDD
scala> val df = spark.read.json("/core_flow/spark/json/001/people.json")df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]                

scala> 

scala> val dfToRDD = df.rdd

dfToRDD: org.apache.spark.rdd.RDD[org.apache.spark.sql.Row] = MapPartitionsRDD[6] at rdd at <console>:29

scala>

scala> dfToRDD.collect

res0: Array[org.apache.spark.sql.Row] = Array([null,Michael], [30,Andy], [19,Justin])

scala>
1.4.2.3 DataSet
1.4.2.3.1 创建
scala> case class Person(name: String, age: Long)

defined class Person

scala> 

scala> val caseClassDS = Seq(Person("Andy", 32)).toDS()

caseClassDS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

scala> 

scala> caseClassDS.show

+----+---+

|name|age|

+----+---+

|Andy| 32|

+----+---+



scala>
1.4.2.3.2 RDD转换为DataSet
scala> val peopleRDD = sc.textFile("examples/src/main/resources/people.txt")

peopleRDD: org.apache.spark.rdd.RDD[String] = examples/src/main/resources/people.txt MapPartitionsRDD[8] at textFile at <console>:28



scala>

scala> case class Person(name: String, age: Long)

defined class Person

scala> 

scala> peopleRDD.map(line => {val para = line.split(",");Person(para(0),para(1).trim.toInt)}).toDS

res2: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

scala>
1.4.2.3.3 DataSet转换为RDD
scala> val DS= Seq(Person("Andy", 32)).toDS()

DS: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

scala> 

scala> DS.rdd

res3: org.apache.spark.rdd.RDD[Person] = MapPartitionsRDD[12] at rdd at <console>:28



scala> res3.collect

res4: Array[Person] = Array(Person(Andy,32))

scala>
1.4.2.4 DataFrame与DataSet相互操作
1.4.2.4.1 DataFrame转Dataset
scala> val df = spark.read.json("./examples/src/main/resources/people.json")

scala> case class Person(name: String, age: Long)

defined class Person

scala> 

scala> df.as[Person]

res14:  org.apache.spark.sql.Dataset[Person]  =  [age:  bigint,  name: string]

scala>
1.4.2.4.2 Dataset转DataFrame
scala> case class Person(name: String, age: Long)

defined class Person

scala>

scala> val ds = Seq(Person("Andy", 32)).toDS()

ds: org.apache.spark.sql.Dataset[Person] = [name: string, age: bigint]

scala> 

scala> val df = ds.toDF

df: org.apache.spark.sql.DataFrame = [name: string, age: bigint]



scala> df.show

+----+---+

|name|age|

+----+---+

|Andy| 32|

+----+---+

scala>
1.4.2.5 RDD / DataFrame / DataSet

enter image description here

在SparkSQL中Spark为提供了两个新抽象,分别是DataFrame和DataSet.
他们和RDD有什么区别? 首先从版本的产生上来看 : 

RDD (Spark1.0) —> Dataframe(Spark1.3) —> Dataset(Spark1.6)

如果同样数据都给到这三个数据结构,他们分别计算之后,都会给出相同结果,不同是执行效率和执行方式.
在后期Spark版本中,DataSet会逐步取代RDD和DataFrame成为唯一的API接口.

1.4.2.5.1 三者共性
DF.map{

 caseRow(col1:String,col2:Int)=>

  println(col1);println(col2)

   col1

    case_=> ""

}

// 定义字段名和类型

caseclassColtest(col1:String,col2:Int)extendsSerializable

DS.map{

 caseColtest(col1:String,col2:Int)=>

  println(col1);println(col2)

   col1

    case_=> ""

}
1.4.2.5.2 三者区别
DF.foreach{

 line=>

  valcol1=line.getAs[String]("col1")

  valcol2=line.getAs[String]("col2")

}
1.4.2.6 SparkSQL Application
package com.geekparkhub.core.spark.application.sparksql



import org.apache.spark.sql.{DataFrame, SparkSession}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * SqlAction

  * <p>

  */



object SqlAction {

  def main(args: Array[String]): Unit = {

    // 创建SparkSession

    val sparkSession: SparkSession = SparkSession

      .builder().master("local[*]").appName("SqlAction").getOrCreate()



    // 导入隐式转换

    import sparkSession.implicits._



    // 创建DF

    val df: DataFrame = sparkSession.read.json("/Volumes/GEEK-SYSTEM/Technical_Framework/spark/projects/spark_server/spark-sql/data/people.json")



    // SQL风格 数据查询 | 创建临时表

    df.createTempView("PEOPLE")

    sparkSession.sql("SELECT * FROM PEOPLE").show()



    // DSL风格 数据查询

    df.select("name").show()



    // 关闭资源

    sparkSession.stop()

  }

}
1.4.2.7 自定义函数
1.4.2.7.1 自定义UDF函数
scala> val df = spark.read.json("hdfs://systemhub511:9000/core_flow/spark/json/001/people.json")

df: org.apache.spark.sql.DataFrame = [age: bigint, name: string]



scala> df.show()

+----+-------+

| age|   name|

+----+-------+

|null|Michael|

|  30|   Andy|

|  19| Justin|

+----+-------+



scala>

scala> spark.udf.register("addName",(x:String) => "Name:" + x)

res1: org.apache.spark.sql.expressions.UserDefinedFunction = UserDefinedFunction(<function1>,StringType,Some(List(StringType)))



scala>

scala> df.createOrReplaceTempView("people")

scala> spark.sql("Select addName(name),age from people").show()

+-----------------+----+

|UDF:addName(name)| age|

+-----------------+----+

|     Name:Michael|null|

|        Name:Andy|  30|

|      Name:Justin|  19|

+-----------------+----+

scala>
1.4.2.7.2 自定义聚合函数
package com.geekparkhub.core.spark.application.aggregation



import org.apache.spark.sql.Row

import org.apache.spark.sql.expressions.{MutableAggregationBuffer, UserDefinedAggregateFunction}

import org.apache.spark.sql.types.{DataType, DoubleType, LongType, StructField, StructType}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * AvgAction

  * <p>

  */



object AvgAction extends UserDefinedAggregateFunction {



  // 定义输入数据类型

  override def inputSchema: StructType = StructType(StructField("input", LongType) :: Nil)



  // 缓存中间值类型

  override def bufferSchema: StructType = StructType(StructField("sum", LongType) :: StructField("count", LongType) :: Nil)



  // 定义输出数据类型

  override def dataType: DataType = DoubleType



  // 函数稳定参数

  override def deterministic: Boolean = true



  // 初始化缓存数据

  override def initialize(buffer: MutableAggregationBuffer): Unit = {

    buffer(0) = 0L

    buffer(1) = 0L

  }



  // 在执行器之内更新

  override def update(buffer: MutableAggregationBuffer, input: Row): Unit = {

    buffer(0) = buffer.getLong(0) + input.getLong(0)

    buffer(1) = buffer.getLong(1) + 1L

  }



  // 在执行器之外合并

  override def merge(buffer1: MutableAggregationBuffer, buffer2: Row): Unit = {

    buffer1(0) = buffer1.getLong(0) + buffer2.getLong(0)

    buffer1(1) = buffer1.getLong(1) + buffer2.getLong(1)



  }



  // 执行数据计算

  override def evaluate(buffer: Row): Any = buffer.getLong(0).toDouble / buffer.getLong(1)

}

package com.geekparkhub.core.spark.application.aggregation



import org.apache.spark.sql.{DataFrame, SparkSession}



/**

  * Geek International Park | 极客国际公园

  * GeekParkHub | 极客实验室

  * Website | https://www.geekparkhub.com/

  * Description | Open开放 · Creation创想 | OpenSource开放成就梦想 GeekParkHub共建前所未见

  * HackerParkHub | 黑客公园枢纽

  * Website | https://www.hackerparkhub.org/

  * Description | 以无所畏惧的探索精神 开创未知技术与对技术的崇拜

  * GeekDeveloper : JEEP-711

  *

  * @author system

  * <p>

  * UdafAction

  * <p>

  */



object UdafAction {

  def main(args: Array[String]): Unit = {



    // 创建SparkSession

    val sparkSession: SparkSession = SparkSession

      .builder().master("local[*]").appName("UdafAction").getOrCreate()



    // 创建DF

    val df: DataFrame = sparkSession.read.json("/Volumes/GEEK-SYSTEM/Technical_Framework/spark/projects/spark_server/spark-sql/data/people.json")



    // SQL风格 数据查询 | 创建临时表

    df.createTempView("PEOPLE")



    // 注册自定义函数

    sparkSession.udf.register("AvgAction", AvgAction)



    // 使用自定义函数

    sparkSession.sql("SELECT AvgAction(age) FROM PEOPLE").show()



    // 关闭资源

    sparkSession.stop()

  }

}

🔒 尚未解锁 正在探索中… 尽情期待 Blog更新! 🔒

1.4.3 Spark SQL 数据源

1.4.3.1 通用加载 / 保存方法
1.4.3.2 JSON文件
1.4.3.3 Parquet文件
1.4.3.4 JDBC
1.4.3.5 Hive DataBase

1.4.4 Spark SQL 实例

🔥 1.5 Spark Streaming 🔥

1.5.1 Spark Streaming 概述

1.5.2 Spark Streaming Program

1.5.3 DataStream 概述

1.5.4 DataStream 输入

1.5.5 DataStream 转换

1.5.6 DataStream 输出

1.5.7 7*24hour运行

1.5.8 Spark Streaming 实例

🔥 2. Spark 高阶 🔥

2.1 内核机制

2.1 性能调优

3. 修仙之道 技术架构迭代 登峰造极之势

Alt text

💡如何对该开源文档进行贡献💡

  1. Blog内容大多是手敲,所以难免会有笔误,你可以帮我找错别字。

  2. 很多知识点我可能没有涉及到,所以你可以对其他知识点进行补充。

  3. 现有的知识点难免存在不完善或者错误,所以你可以对已有知识点的修改/补充。

  4. 💡欢迎贡献各领域开源野生Blog&笔记&文章&片段&分享&创想&OpenSource Project&Code&Code Review

  5. 🙈🙈🙈🙈🙈🙈🙈🙈🙈🙈🙈 issues: geekparkhub.github.io/issues 🙈🙈🙈🙈🙈🙈🙈🙈🙈🙈🙈

希望每一篇文章都能够对读者们提供帮助与提升,这乃是每一位笔者的初衷

💌感谢您的阅读 欢迎您的留言与建议💌

捐助 项目的发展离不开你的支持,请开发者喝杯☕Coffee☕吧!

enter image description here

致谢：

捐助时请备注 UserName

ID UserName Donation Money Consume
1 Object WeChatPay 5RMB 一杯可乐
2 泰迪熊看月亮 AliPay 20RMB 一杯咖啡
3 修仙道长 WeChatPay 10RMB 两杯可乐

License 开源协议

Apache License Version 2.0